package org.jeecg.modules.cascv.rabbitmq;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.jeecg.modules.cascv.image.service.impl.MergeImageService;
import org.jeecg.modules.cascv.image.service.impl.MergeVideoService;
import org.jeecg.modules.cascv.image.util.LaptopConstant;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;

@Component
@Slf4j
public class RabbimtMQListener {
    @Autowired
    private MergeImageService mergeImageService;
    @Autowired
    private MergeVideoService mergeVideoService;

    private List<String> jsonDataList = new CopyOnWriteArrayList<>();


    @RabbitListener(queues = "image_queue")
    public void listenerQueue(Message message){
       // System.out.println(new String(message.getBody()));

        if (message == null) {
            log.error("RabbitMQ消息内容为空！");
            return;
        }
        String jsonStr = null;

        try {
            jsonStr = new String(message.getBody(), "utf-8");

            jsonDataList.add(jsonStr);
            JSONObject jsonObject = JSONObject.parseObject(jsonStr);
            //log.info("接收数据中==>"+jsonStr.substring(14,23)); //idsname
           // log.info("数据："+jsonStr);
          //  num2.incrementAndGet();
          //  log.info("获得条数："+num2.get());  //11929   一秒大概收到30条数据 11929  352    一秒2张图    22个图片 10281  2.2
            //无容器 6D7D  89981总条数 3.24.506     开始52.12.768   684506

            mergeImageService.mergeImage(jsonObject, null);
        } catch (Exception e) {
            log.error("", e);
        }

    }

        @RabbitListener(queues = "video_queue")
    public void videoListener(Message message) {

        if (message == null) {
            log.error("kafka消息内容为空！");
            return;
        }
        String jsonStr = null;
        try {
            jsonStr = new String(message.getBody(), "utf-8");
            jsonDataList.add(jsonStr);
           // LaptopConstant.jsonDataList.add(jsonStr);
            JSONObject jsonObject = JSONObject.parseObject(jsonStr);

            mergeVideoService.merge(jsonObject,null, null);
        } catch (Exception e) {
            log.error("", e);
        }
    }

    public List<String> getJsonDataList() {
        return jsonDataList;
    }






}




/*
prefetch默认值以前是1，这可能会导致高效使用者的利用率不足。从spring-amqp 2.0版开始，默认的prefetch值是250，这将使消费者在大多数常见场景中保持忙碌，从而提高吞吐量。
不过在有些情况下，尤其是处理速度比较慢的大消息，消息可能在内存中大量堆积，消耗大量内存；以及对于一些严格要求顺序的消息，prefetch的值应当设置为1。
对于低容量消息和多个消费者的情况（也包括单listener容器的concurrency配置）希望在多个使用者之间实现更均匀的消息分布，建议在手动ack下并设置prefetch=1
 */